package net.wicp.tams.common.es.client;

import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.sql.Connection;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

import org.apache.commons.beanutils.PropertyUtils;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.Validate;
import org.apache.http.Header;
import org.apache.http.HttpEntity;
import org.apache.http.HttpHost;
import org.apache.http.client.config.RequestConfig.Builder;
import org.apache.http.entity.ContentType;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.apache.http.message.BasicHeader;
import org.apache.http.nio.entity.NStringEntity;
import org.elasticsearch.ElasticsearchStatusException;
import org.elasticsearch.action.ActionFuture;
import org.elasticsearch.action.admin.cluster.settings.ClusterUpdateSettingsRequest;
import org.elasticsearch.action.admin.cluster.settings.ClusterUpdateSettingsResponse;
import org.elasticsearch.action.admin.indices.alias.Alias;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
import org.elasticsearch.action.admin.indices.flush.FlushRequest;
import org.elasticsearch.action.admin.indices.flush.FlushResponse;
import org.elasticsearch.action.admin.indices.get.GetIndexRequest;
import org.elasticsearch.action.admin.indices.get.GetIndexResponse;
import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
import org.elasticsearch.action.admin.indices.refresh.RefreshResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.IndicesAdminClient;
import org.elasticsearch.client.Response;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.cluster.metadata.MappingMetaData;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.TransportAddress;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.TermQueryBuilder;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.elasticsearch.transport.client.PreBuiltTransportClient;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializeConfig;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.carrotsearch.hppc.cursors.ObjectObjectCursor;

import lombok.extern.slf4j.Slf4j;
import net.wicp.tams.common.Conf;
import net.wicp.tams.common.Result;
import net.wicp.tams.common.apiext.CollectionUtil;
import net.wicp.tams.common.apiext.StringUtil;
import net.wicp.tams.common.apiext.json.JSONUtil;
import net.wicp.tams.common.constant.RestMethod;
import net.wicp.tams.common.constant.dic.YesOrNo;
import net.wicp.tams.common.es.EsAssit;
import net.wicp.tams.common.es.EsData;
import net.wicp.tams.common.es.MappingEnumSerializer;
import net.wicp.tams.common.es.bean.AliasesBean;
import net.wicp.tams.common.es.bean.IndexBean;
import net.wicp.tams.common.es.bean.IndexParamsBean;
import net.wicp.tams.common.es.bean.MappingBean;
import net.wicp.tams.common.es.bean.MappingBean.DataTypes;
import net.wicp.tams.common.es.bean.MappingBean.Dynamic;
import net.wicp.tams.common.es.bean.MappingBean.Propertie;
import net.wicp.tams.common.es.bean.QueryDataMap;
import net.wicp.tams.common.es.bean.SettingsBean;
import net.wicp.tams.common.es.constant.ClusterSettings;
import net.wicp.tams.common.es.constant.ClusterSettings.Type;
import net.wicp.tams.common.es.norm.IQueryData;
import net.wicp.tams.common.exception.ExceptAll;
import net.wicp.tams.common.exception.ProjectExceptionRuntime;
import net.wicp.tams.common.http.HttpResult;
import net.wicp.tams.common.web.PageAssist;

@Slf4j
public class ESClient {

	private RestClient restClient;
	private RestHighLevelClient restHighLevelClient;
	private TransportClient client;

	public ESClient(final Properties props) {
		init(props == null ? Conf.copyProperties() : props);
	}

	public ESClient() {
		this(null);
	}

	@SuppressWarnings("resource")
	private ESClient init(final Properties props) {
		String hoststrs = props.getProperty("common.es.host.name");
		String[] hostary = hoststrs.split(",");
		HttpHost[] httpHosts = new HttpHost[hostary.length];
		for (int i = 0; i < httpHosts.length; i++) {
			httpHosts[i] = new HttpHost(hostary[i], Integer.parseInt(props.getProperty("common.es.host.port.rest")),
					props.getProperty("common.es.host.scheme"));
		}
		RestClientBuilder builder = RestClient.builder(httpHosts);
		if (StringUtil.isNotNull(props.getProperty("common.es.http.head"))) {
			String[] eles = props.getProperty("common.es.http.head").split("&", 0);
			Header[] defaultHeaders = new Header[eles.length];
			for (int i = 0; i < defaultHeaders.length; i++) {
				String[] split = eles[i].split("=");
				defaultHeaders[i] = new BasicHeader(split[0], split[1]);
			}
			builder.setDefaultHeaders(defaultHeaders);
		}

		if (Integer.parseInt(props.getProperty("common.es.http.maxRetryTimeoutMillis")) > 0) {
			builder.setMaxRetryTimeoutMillis(
					Integer.parseInt(props.getProperty("common.es.http.maxRetryTimeoutMillis")));
		}
		// 异步httpclient的连接延时配置
		builder.setRequestConfigCallback(new RestClientBuilder.RequestConfigCallback() {
			@Override
			public Builder customizeRequestConfig(Builder requestConfigBuilder) {
				requestConfigBuilder
						.setConnectTimeout(Integer.parseInt(props.getProperty("common.es.http.connectTimeout")));
				requestConfigBuilder
						.setSocketTimeout(Integer.parseInt(props.getProperty("common.es.http.socketTimeout")));
				requestConfigBuilder.setConnectionRequestTimeout(
						Integer.parseInt(props.getProperty("common.es.http.connectionRequestTimeout")));
				return requestConfigBuilder;
			}
		});

		// 异步httpclient的连接数配置
		builder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
			@Override
			public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
				httpClientBuilder
						.setMaxConnTotal(Integer.parseInt(props.getProperty("common.es.http.aync.maxConnTotal")));
				httpClientBuilder
						.setMaxConnPerRoute(Integer.parseInt(props.getProperty("common.es.http.aync.maxConnPerRoute")));
				return httpClientBuilder;
			}
		});
		restHighLevelClient = new RestHighLevelClient(builder);
		restClient = restHighLevelClient.getLowLevelClient();

		System.setProperty("es.set.netty.runtime.available.processors", "false");
		try {
			// 设置集群名称
			Settings.Builder settingsBuilder = Settings.builder();
			settingsBuilder.put("client.transport.sniff", true);
			String custerName = StringUtil.hasNull(props.getProperty("common.es.cluster.name"), "elasticsearch");
			settingsBuilder.put("cluster.name", custerName);// .build();// 集群名
			Map<String, String> pres = CollectionUtil.getPropsByKeypre(props, "common.es.client", true);
			for (String clientkey : pres.keySet()) {
				settingsBuilder.put("client." + clientkey, pres.get(clientkey));
			}
			// 创建client
			TransportAddress[] addrs = new TransportAddress[hostary.length];
			for (int i = 0; i < addrs.length; i++) {
				addrs[i] = new TransportAddress(InetAddress.getByName(hostary[i]),
						Integer.parseInt(props.getProperty("common.es.host.port.transport")));
			}
			client = new PreBuiltTransportClient(settingsBuilder.build()).addTransportAddresses(addrs);
		} catch (UnknownHostException e) {
			log.error("创建client失败", e);
			throw new RuntimeException("创建client失败", e);
		}
		log.info("创建ESClient成功");
		return this;
	}

	//////////////////////////////////////////////////////////////////////////
	public ClusterUpdateSettingsResponse admin_ClusterSettings(Map<ClusterSettings, String> settingmap) {
		ClusterUpdateSettingsRequest request = new ClusterUpdateSettingsRequest();
		Settings.Builder builderTransient = Settings.builder();// .put("indices.store.throttle.type", "none").build();
		Settings.Builder builderPersistent = Settings.builder();
		for (ClusterSettings set : settingmap.keySet()) {
			set.valid(settingmap.get(set));
			if (set.getType() == net.wicp.tams.common.es.constant.ClusterSettings.Type.PERSISTENT) {
				builderPersistent.put(set.getKey(), settingmap.get(set));
			}
			if (set.getType() == Type.TRANSIENT) {
				builderTransient.put(set.getKey(), settingmap.get(set));
			}
		}
		if (CollectionUtils.isNotEmpty(builderTransient.keys())) {
			request.transientSettings(builderTransient.build());
		}
		if (CollectionUtils.isNotEmpty(builderPersistent.keys())) {
			request.persistentSettings(builderPersistent.build());
		}
		ActionFuture<ClusterUpdateSettingsResponse> updateSettings = client.admin().cluster().updateSettings(request);
		ClusterUpdateSettingsResponse clusterUpdateSettingsResponse = null;
		try {
			clusterUpdateSettingsResponse = updateSettings.get();
			return clusterUpdateSettingsResponse;
		} catch (Exception e) {
			log.error("更新群集配置错误", e);
			throw new RuntimeException("更新群集配置错误");
		}
	}

	public RestClient getRestClient() {
		return restClient;
	}

	public RestHighLevelClient getRestHighLevelClient() {
		return restHighLevelClient;
	}

	public void close() {
		if (this.restClient != null) {
			try {
				restClient.close();
				restHighLevelClient.close();
			} catch (IOException e) {
				log.error("关闭ES链接失败", e);
			}
		}
		if (this.client != null) {
			try {
				this.client.close();
			} catch (Exception e) {
				log.error("关闭tcp链接失败", e);
			}
		}
	}

	/****
	 * 一般对象的批量新增
	 * 
	 * @param index
	 * @param type
	 * @param idColName  id字段的列名
	 * @param insertList
	 * @return
	 */
	public <T> Result docWriteBatch(String index, String type, String idColName, List<T> insertList) {
		BulkRequest bulkRequest = new BulkRequest();
		for (T insertObj : insertList) {
			IndexRequest indexRequest = new IndexRequest(index, type);
			if (StringUtil.isNotNull(idColName)) {
				try {
					Object property = PropertyUtils.getProperty(insertObj, idColName);
					indexRequest.id(String.valueOf(property));
				} catch (Exception e) {
					return Result.getError(e.getMessage());
				}
			}
			indexRequest.source(JSON.toJSONString(insertObj), XContentType.JSON);
			bulkRequest.add(indexRequest);
		}
		try {
			this.restHighLevelClient.bulk(bulkRequest);
			return Result.getSuc();
		} catch (IOException e) {
			log.error("批量添加文档错误", e);
			return Result.getError("批量添加文档错误" + e.getMessage());
		}
	}

	// 序列化过来的数据
	public Result docWriteBatch(List<EsData> esDatas) {
		BulkRequest bulkRequest = new BulkRequest();
		for (EsData esData : esDatas) {
			Map<String, DataTypes> mapping = queryMapping_tc(esData.getIndex(), esData.getType());
			bulkRequest.add(EsAssit.packRequest(esData, mapping));
		}
		try {
			BulkResponse bulk = this.restHighLevelClient.bulk(bulkRequest);
			if (bulk.hasFailures()) {
				Result error = new Result(ExceptAll.project_haveerrors);
				error.setRetObjs(bulk.getItems());
				return error;
			} else {
				return Result.getSuc();
			}
		} catch (IOException e) {
			log.error("批量添加文档错误", e);
			return Result.getError("批量添加文档错误" + e.getMessage());
		}
	}

	private final Map<String, Map<String, DataTypes>> mappingMap = new HashMap<>();
	private final String mappingMapKeyFormate = "%s.%s";

	public Result docWriteBatch_tc(List<EsData> esDatas) {
		BulkRequest bulkRequest = new BulkRequest();
		// bulkRequest.setRefreshPolicy(RefreshPolicy.IMMEDIATE);//TODO
		for (EsData esData : esDatas) {
			String key = String.format(mappingMapKeyFormate, esData.getIndex(), esData.getType());
			Map<String, DataTypes> mapping;
			if (mappingMap.get(key) == null) {
				mapping = queryMapping_tc(esData.getIndex(), esData.getType());
				mappingMap.put(key, mapping);
			} else {
				mapping = mappingMap.get(key);
			}
			bulkRequest.add(EsAssit.packRequest(esData, mapping));
		}
		if (bulkRequest.numberOfActions() == 0) {
			log.error("出现数据被过滤");
			return Result.getSuc();
		}
		try {
			BulkResponse bulk = this.client.bulk(bulkRequest).get();
			if (bulk.hasFailures()) {
				Result error = new Result(ExceptAll.project_haveerrors);
				error.setRetObjs(bulk.getItems());
				return error;
			} else {
				return Result.getSuc();
			}
		} catch (Exception e) {
			log.error("批量添加文档错误", e);
			return Result.getError("批量添加文档错误" + e.getMessage());
		}
	}

	/***
	 * 立马refresh索引
	 * 
	 * @param indexName
	 * @return
	 */
	public Result indexRefresh(String... indexName) {
		Result ret = Result.getSuc();
		try {
			RefreshResponse response = client.admin().indices().refresh(new RefreshRequest(indexName)).actionGet();

			if (response.getShardFailures().length == response.getTotalShards()) {
				log.error("ES索引刷新失败" + response.getShardFailures());
				ret = Result.getError("ES索引刷新失败" + response.getShardFailures());
			} else if (response.getShardFailures().length > 0) {
				log.error("ES索引刷新部分分片失败" + response.getShardFailures());
				ret = Result.getError("ES索引刷新部分分片失败" + response.getShardFailures());
			}
		} catch (Exception e) {
			log.error("ES刷新失败", e);
			ret = new Result(ExceptAll.duckula_es_refresh);
		}
		return ret;
	}

	/***
	 * 强制flush数据
	 * 
	 * @param indexName
	 * @return
	 */
	public Result indexFlush(String... indexName) {
		Result ret = Result.getSuc();
		try {
			FlushResponse response = client.admin().indices().flush(new FlushRequest(indexName)).actionGet();
			// 输出json格式的响应信息
			if (response.getShardFailures().length == response.getTotalShards()) {
				log.error("ES索引刷新失败" + response.getShardFailures());
				ret = Result.getError("ES索引刷新失败" + response.getShardFailures());
			} else if (response.getShardFailures().length > 0) {
				log.error("ES索引刷新部分分片失败" + response.getShardFailures());
				ret = Result.getError("ES索引刷新部分分片失败" + response.getShardFailures());
			}
		} catch (Exception e) {
			log.error("ES刷新失败", e);
			ret = new Result(ExceptAll.duckula_es_flush);
		}
		return ret;
	}

	public void cleanMappingMap() {
		this.mappingMap.clear();
	}

	public Result docWriteBatch_tc(EsData... esDatas) {
		return docWriteBatch_tc(Arrays.asList(esDatas));
	}

	@SuppressWarnings("unused")
	public Result indexCreate(String indexName, String type, int shardsNum, int replicas, String alias,
			MappingBean mappingBean) {
		Validate.notEmpty(indexName);
		Validate.isTrue(shardsNum > 0);
		Validate.isTrue(replicas >= 0);
		Validate.notNull(mappingBean);
		CreateIndexRequest request = new CreateIndexRequest(indexName);
		request.settings(
				Settings.builder().put("index.number_of_shards", shardsNum).put("index.number_of_replicas", replicas));
		// 不格式化会报错：java.lang.IllegalArgumentException
		String jsonString2 = mappingFormat(type, mappingBean);
		request.mapping(type, jsonString2, XContentType.JSON);
		if (StringUtil.isNotNull(alias)) {
			request.alias(new Alias(alias));
		}
		try {
			CreateIndexResponse createIndexResponse = this.restHighLevelClient.indices().create(request);// 同步创建索引
			return Result.getSuc();
		} catch (IOException e) {
			log.error("创建索引失败", e);
			return Result.getError("创建索引失败:" + e.getMessage());
		} catch (ElasticsearchStatusException e) {
			log.error("创建索引失败", e);
			return Result.getError("创建索引失败:" + e.getDetailedMessage());
		}
	}

	public Result indexCreate(String indexName, String type, int shardsNum, int replicas, MappingBean mappingBean) {
		return indexCreate(indexName, type, shardsNum, replicas, null, mappingBean);
	}

	public Result indexCreate(String indexName, String type, int shardsNum, MappingBean mappingBean) {
		return indexCreate(indexName, type, shardsNum, 0, null, mappingBean);
	}

	public Result indexCreate(Connection conn, IndexParamsBean mapping, String indexName, int shardsNum, int replicas) {
		MappingBean proMappingBean = EsAssit.proMappingBean(conn, mapping);
		if (proMappingBean == null) {
			return Result.getError("得到MappingBean失败");
		}
		return indexCreate(indexName, "_doc", shardsNum, 0, null, proMappingBean);
	}

	public Result aliasCreate(String indexNamePatten, String... aliass) {
		JSONArray actions = new JSONArray();
		for (String alias : aliass) {
			actions.add(JSON.parseObject(
					" { \"add\" : { \"index\" : \"" + indexNamePatten + "\", \"alias\" : \"" + alias + "\" } }"));
		}
		JSONObject params = JSONUtil.packParams("actions", actions);
		HttpResult callRest = callRest(RestMethod.POST, "/_aliases", params);
		return callRest.getResult();
	}

	/***
	 * 替换别名
	 * 
	 * @param indexNew
	 * @param indexOld
	 * @param idDelOld 是否删除旧索引，用于交换别名时用
	 * @param aliass
	 * @return
	 */
	public Result indexReplace(String indexNew, String indexOld, boolean idDelOld, String... aliass) {
		JSONArray actions = new JSONArray();
		for (String alias : aliass) {
			actions.add(JSON
					.parseObject(" { \"add\" : { \"index\" : \"" + indexNew + "\", \"alias\" : \"" + alias + "\" } }"));
			if (idDelOld) {
				actions.add(JSON.parseObject(" { \"remove_index\": { \"index\": \"" + indexOld + "\" } }"));
			} else {
				actions.add(JSON.parseObject(
						" { \"remove\" : { \"index\" : \"" + indexOld + "\", \"alias\" : \"" + alias + "\" } }"));
			}
		}
		JSONObject params = JSONUtil.packParams("actions", actions);
		HttpResult callRest = callRest(RestMethod.POST, "/_aliases", params);
		return callRest.getResult();
	}

	public Result indexDel(String index) {
		JSONArray actions = new JSONArray();
		actions.add(JSON.parseObject(" { \"remove_index\": { \"index\": \"" + index + "\" } }"));
		JSONObject params = JSONUtil.packParams("actions", actions);
		HttpResult callRest = callRest(RestMethod.POST, "/_aliases", params);
		return callRest.getResult();
	}

	/**
	 * 设置refresh和副本数量
	 * 
	 * @param index
	 * @param settingsBean
	 * @return
	 */
	public Result indexSetting(String index, SettingsBean settingsBean) {
		JSONObject obj = JSON.parseObject(JSONObject.toJSONString(settingsBean));
		Validate.isTrue(obj.size() > 0, "副本数与刷新间隔必须填一个");
		if (SettingsBean.fresh_null.equals(settingsBean.getRefresh_interval())) {
			obj.put("refresh_interval", null);
		}
		String jsonString = JSON.toJSONString(obj, SerializerFeature.WriteMapNullValue);
		String point = String.format("/%s/_settings", index);
		HttpResult callRest = callRest(RestMethod.PUT, point, String.format("{\"index\" : %s}", jsonString));
		return callRest.getResult();
	}

	public HttpResult callRest(RestMethod restMethod, String point, JSONObject params) {
		return callRest(restMethod, point, params.toJSONString());
	}

	public HttpResult callRest(RestMethod restMethod, String point, String paramsStr) {
		HttpEntity entity = null;
		if (StringUtil.isNotNull(paramsStr)) {
			entity = new NStringEntity(paramsStr, ContentType.APPLICATION_JSON);
		}
		try {
			Response response = restClient.performRequest(restMethod.name(), point,
					Collections.<String, String>emptyMap(), entity);

			HttpResult httpResult = EsClientAssit.packHttpResult(response);
			return httpResult;
		} catch (IOException e) {
			log.error("rest形式调用失败", e);
			HttpResult ret = new HttpResult();
			ret.setErrMsg(e.getMessage());
			return ret;
		}
	}

	private String mappingFormat(String type, MappingBean mappingBean) {
		SerializeConfig tamsSerializeConfig = SerializeConfig.getGlobalInstance();
		tamsSerializeConfig.put(DataTypes.class, MappingEnumSerializer.instance);
		tamsSerializeConfig.put(Dynamic.class, MappingEnumSerializer.instance);
		String jsonString = JSON.toJSONString(JSONUtil.packParams(type, mappingBean), tamsSerializeConfig);
		String jsonString2 = JSON.toJSONString(JSON.parse(jsonString), true);
		return jsonString2;
	}

	public Result docWrite(String index, String type, Object data) {
		IndexRequest indexRequest = new IndexRequest(index, type);
		indexRequest.source(JSON.toJSONString(data), XContentType.JSON);
		try {
			this.restHighLevelClient.index(indexRequest);
			return Result.getSuc();
		} catch (IOException e) {
			log.error("添加文档错误", e);
			return Result.getError("添加文档错误" + e.getMessage());
		}
	}

	public <T> Result docUpdate(String index, String type, String id, T updateObj) {
		UpdateRequest updateRequest = new UpdateRequest(index, type, id);
		updateRequest.doc(JSON.toJSONString(updateObj), XContentType.JSON);
		try {
			this.restHighLevelClient.update(updateRequest);
			return Result.getSuc();
		} catch (IOException e) {
			log.error("更新文档错误", e);
			return Result.getError("更新文档错误" + e.getMessage());
		}
	}

	public Result docDel(String index, String type, String id) {
		DeleteRequest deleteRequest = new DeleteRequest(index, type, id);
		try {
			this.restHighLevelClient.delete(deleteRequest);
			return Result.getSuc();
		} catch (IOException e) {
			log.error("删除文档错误", e);
			return Result.getError("删除文档错误" + e.getMessage());
		}
	}

	public void docDelBatchQuery(String index, String type, String key, String value) {
		try {
			SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
			sourceBuilder.timeout(new TimeValue(2, TimeUnit.SECONDS));
			TermQueryBuilder termQueryBuilder1 = QueryBuilders.termQuery(key, value);
			sourceBuilder.query(termQueryBuilder1);
			SearchRequest searchRequest = new SearchRequest(index);
			searchRequest.types(type);
			searchRequest.source(sourceBuilder);
			SearchResponse response = this.restHighLevelClient.search(searchRequest);
			SearchHits hits = response.getHits();
			List<String> docIds = new ArrayList<>(hits.getHits().length);
			for (SearchHit hit : hits) {
				docIds.add(hit.getId());
			}
			BulkRequest bulkRequest = new BulkRequest();
			for (String id : docIds) {
				DeleteRequest deleteRequest = new DeleteRequest(index, type, id);
				bulkRequest.add(deleteRequest);
			}
			this.restHighLevelClient.bulk(bulkRequest);
		} catch (IOException e) {
			log.error("批量删除文档错误", e);
			throw new RuntimeException("批量删除文档错误", e);
		}
	}

	public List<IndexBean> queryIndex(String indexPattern) {
		String cmd = StringUtil.isNull(indexPattern) ? "_cat/indices?v" : "_cat/indices/" + indexPattern + "?v&s=index";
		List<IndexBean> retlist = callCat(IndexBean.class, cmd);
		return retlist;
	}

	public List<AliasesBean> queryAliases(String aliasesPattern) {
		String cmd = StringUtil.isNull(aliasesPattern) ? "_cat/aliases?v" : "_cat/aliases/" + aliasesPattern + "?v";
		List<AliasesBean> retlist = callCat(AliasesBean.class, cmd);
		return retlist;
	}

	public <T> List<T> callCat(Class<T> calssz, String cmd) {
		List<T> retlist = new ArrayList<>();
		HttpResult httpResult = callRest(RestMethod.GET, cmd, "");
		if (!httpResult.getResult().isSuc()) {
			return retlist;
		}
		String bodyStr = httpResult.getBodyStr();
		String[] rows = bodyStr.split("\n");
		String[] head = rows[0].split("[\\s]+");
		for (int i = 1; i < rows.length; i++) {
			String[] cols = rows[i].split("[\\s]+");
			JSONObject json = new JSONObject();
			for (int j = 0; j < cols.length; j++) {
				json.put(head[j].replaceAll("\\.", "_"), cols[j]);
			}
			retlist.add(json.toJavaObject(calssz));
		}
		return retlist;
	}

	public GetIndexResponse queryIndex_tc(String indexName, String... types) {
		GetIndexRequest query = new GetIndexRequest();
		query.indices(indexName);
		if (ArrayUtils.isNotEmpty(types)) {
			query.types(types);
		}
		IndicesAdminClient indicesAdminClient = client.admin().indices();
		try {
			GetIndexResponse resp = indicesAdminClient.getIndex(query).get(5000, TimeUnit.MILLISECONDS);
			return resp;
		} catch (Exception e) {
			log.error("查找索引失败", e);
			throw new RuntimeException("查找索引失败");
		}
	}

	@SuppressWarnings("unchecked")
	public Map<String, DataTypes> queryMapping_tc(String indexName, String type) {
		GetIndexResponse queryIndex_tc = queryIndex_tc(indexName, type);
		Iterator<ObjectObjectCursor<String, MappingMetaData>> iterator = queryIndex_tc.getMappings().get(indexName)
				.iterator();
		Map<String, DataTypes> retmap = new HashMap<>();
		if (iterator.hasNext()) {
			ObjectObjectCursor<String, MappingMetaData> next = iterator.next();
			LinkedHashMap<String, LinkedHashMap<String, String>> object = (LinkedHashMap<String, LinkedHashMap<String, String>>) next.value
					.sourceAsMap().get("properties");
			for (String key : object.keySet()) {
				String typestr = object.get(key).get("type");
				retmap.put(key, DataTypes.getDataTypeByName(typestr));
			}
		}
		return retmap;
	}

	@SuppressWarnings("unchecked")
	public Map<String, MappingBean.Propertie> queryMapping_tc_all(String indexName, String type) {
		GetIndexResponse queryIndex_tc = queryIndex_tc(indexName, type);
		Iterator<ObjectObjectCursor<String, MappingMetaData>> iterator = queryIndex_tc.getMappings().get(indexName)
				.iterator();
		Map<String, Propertie> retmap = new HashMap<>();
		if (iterator.hasNext()) {
			ObjectObjectCursor<String, MappingMetaData> next = iterator.next();
			LinkedHashMap<String, LinkedHashMap<String, String>> object = (LinkedHashMap<String, LinkedHashMap<String, String>>) next.value
					.sourceAsMap().get("properties");
			for (String key : object.keySet()) {
				Propertie props = new Propertie();
				String typestr = object.get(key).get("type");
				props.setType(DataTypes.getDataTypeByName(typestr));
				if (object.get(key).containsKey("relations")) {

					String jsonForMap = JSONUtil.getJsonForMap(object.get(key));

					props.setRelations(JSONObject.parseObject(jsonForMap).getJSONObject("relations"));
				}
				retmap.put(key, props);
			}
		}
		return retmap;
	}

	///////////////////////////////////// 查询封装///////////////////////////////////////////////////////////////////////////////////
	private TimeValue timeValue = new TimeValue(Integer.parseInt(Conf.get("common.es.query.timeout")),
			TimeUnit.MILLISECONDS);

	@SuppressWarnings("unchecked")
	public <T extends IQueryData> List<T> queryDocPage(Class<T> classz, PageAssist pageAssist,
			QueryBuilder queryBuilder, String index, String type) {
		/*
		 * SearchRequestBuilder searchRequestBuilder = client.prepareSearch();
		 * searchRequestBuilder.setIndices(index); searchRequestBuilder.setTypes(type);
		 * searchRequestBuilder.setFrom(pageAssist.getFrom());
		 * searchRequestBuilder.setSize(pageAssist.getPageSize());
		 * searchRequestBuilder.setQuery(queryBuilder); SearchResponse response =
		 * searchRequestBuilder.get();
		 */
		index = StringUtil.hasNull(index, Conf.get("common.es.query.index"));
		Validate.isTrue(StringUtil.isNotNull(index) && !"none".equals(index), "需要指定索引");
		type = StringUtil.hasNull(type, Conf.get("common.es.query.type"));
		SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
		sourceBuilder.timeout(timeValue);
		sourceBuilder.query(queryBuilder);
		sourceBuilder.from(pageAssist.getFrom());
		sourceBuilder.size(pageAssist.getPageSize());
		if (StringUtil.isNotNull(pageAssist.getSortField())) {
			sourceBuilder.sort(pageAssist.getSortField(),
					pageAssist.getSortDesc() == YesOrNo.yes ? SortOrder.DESC : SortOrder.ASC);
		}
		SearchRequest searchRequest = new SearchRequest(index);
		searchRequest.types(type);
		searchRequest.source(sourceBuilder);

		SearchResponse response = null;
		try {
			response = this.restHighLevelClient.search(searchRequest);
		} catch (IOException e) {
			throw new ProjectExceptionRuntime(ExceptAll.duckula_es_query);
		}
		SearchHits hits = response.getHits();
		long totalHits = hits.getTotalHits();
		pageAssist.setAllNum(totalHits);
		List<T> retlist = new ArrayList<>();
		for (SearchHit searchHit : hits) {
			System.out.println("searchHit----" + searchHit.getId());
			T parseObject = null;
			if ("net.wicp.tams.common.es.bean.QueryDataMap".equals(classz.getName())) {
				QueryDataMap queryDataMap = new QueryDataMap();
				queryDataMap.setData(searchHit.getSourceAsMap());
				parseObject = (T) queryDataMap;
			} else {
				parseObject = JSONObject.parseObject(searchHit.getSourceAsString(), classz);
			}
			IQueryData querydata = (IQueryData) parseObject;
			querydata.set_id(searchHit.getId());
			querydata.set_index(searchHit.getIndex());
			querydata.set_type(searchHit.getType());
			retlist.add(parseObject);
		}
		pageAssist.setResult(retlist);
		return retlist;
	}

	public <T extends IQueryData> List<T> queryDocPage(Class<T> classz, PageAssist pageAssist,
			QueryBuilder queryBuilder) {
		return queryDocPage(classz, pageAssist, queryBuilder, null, null);
	}

	public TransportClient getClient() {
		return client;
	}
}
